package midas.mw.crt.presence;

import java.util.Enumeration;
import java.util.HashMap;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
import java.util.TimerTask;
import java.util.Vector;


import midas.mw.cfg.CfgError;
import midas.mw.cfg.CfgLogging;
import midas.mw.cfg.log.Logger;
import midas.mw.crt.presence.util.MD5ID;
import net.jxta.util.AdvertisementUtilities;
import net.jxta.peergroup.PeerGroup;
import net.jxta.rendezvous.RendezvousListener;
import net.jxta.rendezvous.RendezvousEvent;
import net.jxta.discovery.DiscoveryService;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.id.ID;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.PipeService;
import net.jxta.pipe.InputPipe;
import net.jxta.pipe.OutputPipe;
import net.jxta.pipe.PipeMsgListener;


import net.jxta.document.AdvertisementFactory;

/**
 * This describes a simple system for detecting when nodes enter and leave the 
 * MIDAS network. How presence works?
 * <p>
 * The Central Node keeps track of nodes connected from the group. When a node
 * enters or leaves a Central Node, the Central Node does two things:
 * <ul>
 * <li>Sends a list of currently connected nodes on a bulk pipe
 * <li>Sends the notication about connections and disconnections on a presence pipe
 * </ul>
 * This updates the new node with the current list and updates all the connected
 * nodes.
 */

public class NodePresence
{    
    /**
     * Buffer for bulk announce.
     */
    protected StringBuffer buff1 = new StringBuffer();
    
    /**
     * Peers in the MIDAS group reporting from all Central Nodes
     */
    //protected HashMap midasNodes = new HashMap();
    
    /**
     * List of local nodes connected to this Central Node
     */
    protected HashMap localNodes = new HashMap();
    
    /**
     * Announce in pipe. Used to get PresenceKey messages about other nodes.
     */
    protected InputPipe announceIn = null;
    
    /**
     * Announce out pipe. Used to send a PresenceKey.
     */
    protected OutputPipe announceOut = null;
    /**
     * Bulk in pipe
     */
    protected InputPipe bulkIn = null;
    
    /**
     * Bulk out pipe
     */
    protected OutputPipe bulkOut = null;
    
    /**
     * The node group context
     */
    protected PeerGroup peerGroup;
    
    /**
     * The jxta ID for the node in the MIDAS group
     */
    protected String thisNodeID;
    
    /**
     * Advertisement for the bulk messages
     */
    protected PipeAdvertisement bulkPipeAdvertisement;
    
    /**
     * List of listeners to presence events
     */
    protected ArrayList presenceListeners = new ArrayList();
    
    /**
     * Data describing the node in the MIDAS group
     */
    protected PresenceAdv localPresenceAdv;
    
    /**
     * Local PresenceKey.
     */
    protected PresenceKey localPresenceKey;
    
    protected boolean headless = false;
    
    /** 
     * Delay before a bulk announce is made. 
     */
    protected long bulkAnnounceDelay = 15*1000L;
    
    /**
     * Creates a new instance of NodePresence
     * @param peerGroup Group context
     * @param localPresenceAdv Data describing the node in this group
     */
    public NodePresence(PeerGroup peerGroup,PresenceAdv localPresenceAdv) 
    {
        this( peerGroup,null,localPresenceAdv);
    }
    
    /**
     * Creates a new instance of NodePresence
     * @param peerGroup Group context.
     * @param listener Listener for presence events
     * @param localPresenceAdv Data describing the node in this group
     */
    public NodePresence(PeerGroup peerGroup, PresenceListener listener,PresenceAdv localPresenceAdv) 
    {    
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence(PeerGroup, PresenceListener, PresenceAdv)");
    	
        this.peerGroup 			= peerGroup;
        this.localPresenceAdv 	= localPresenceAdv;
        thisNodeID 				= peerGroup.getPeerID().toString();
        localPresenceAdv.setPesenceNodeID(thisNodeID);
        String id 				= PresenceAdv.makeID(peerGroup, thisNodeID);
        localPresenceAdv.setID(id);
        //LOG.info("============================== Publishing ========================");
        try{
            peerGroup.getDiscoveryService().publish(localPresenceAdv,
                    DiscoveryService.INFINITE_LIFETIME,
                    DiscoveryService.DEFAULT_LIFETIME);
        }catch(Exception e){
        	CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_ERROR, "NodePresence(PeerGroup, PresenceListener, PresenceAdv). " + 
					"Error pubishing our own advertisement: '" + e.getMessage() + "'", null);            
            throw new RuntimeException("Critical Error publishing our own advertisement");
        }
        
        localPresenceKey =  new PresenceKey(thisNodeID,localPresenceAdv.getPresenceName(),localPresenceAdv.getNodeType(),localPresenceAdv.getProxyName(),localPresenceAdv.getLastChanged());
        if(listener!= null)
            addPresenceListener( listener);
        
        //midasNodes.put(localPresenceKey.getPresenceNodeID(),localPresenceKey);
        localNodes.put(localPresenceKey.getPresenceNodeID(),localPresenceKey);
        peerGroup.getRendezVousService().addListener(new RDVListener());
        ID bulkPipeID 			= (net.jxta.id.ID)MD5ID.createPipeID(peerGroup.getPeerGroupID(),"RDV bulk", "status pipe");        
        String type 			= PipeService.PropagateType;
        bulkPipeAdvertisement 	= createPipeAdvertisement(bulkPipeID,type);
        try {
        	peerGroup.getDiscoveryService().publish(bulkPipeAdvertisement);
        	peerGroup.getDiscoveryService().remotePublish(bulkPipeAdvertisement);
        } catch(Exception e){
        	CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_ERROR, "NodePresence(PeerGroup, PresenceListener, PresenceAdv). " + 
					"Error publishing our own advertisement: '" + e.getMessage() + "'", null);            
            throw new RuntimeException("Critical Error publishing our own advertisement");
        }
        startBulkPipeIn();
        startBulkPipeOut();
        startAnnouncePipe();
       
        if (peerGroup.getRendezVousService().isRendezVous())
            new addNodeNow(localPresenceKey).start();
        else
            java.awt.EventQueue.invokeLater(new addNodeNow(localPresenceKey));
            //new addNodeNow(localPresenceKey).run();
    }
    
    static final PipeAdvertisement createPipeAdvertisement(ID pipeID,String type)
    {
    	 CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.createPipeAdvertisement(ID, " + type);
        
    	 PipeAdvertisement pipeAdvertisement = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(PipeAdvertisement.getAdvertisementType());
    	 pipeAdvertisement.setPipeID(pipeID);
    	 pipeAdvertisement.setType(type);
    	 pipeAdvertisement.setName("BulkPipe");
    	 return pipeAdvertisement;
    }
   
    /** 
     * The node announces itself 
    */
    class addNodeNow extends Thread
    {    	    
        PresenceAdv localPresenceAdv;
        addNodeNow(PresenceKey localPresenceKey)
        {
        	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "addNodeNow(PresenceKey)");
        	
            this.localPresenceAdv =localPresenceAdv;
        }
        public void run()
        {
        	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "addNodeNow.run()");
        	
            addPeer(localPresenceKey);
        }
    }
    
    /**
     * Add a presence listener.
     * Does a bulk announce for new listener if there are any nodes already here.
     * @param listener The listener to add.
     */
    public void addPresenceListener(PresenceListener listener)
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.addPresenceListener(PresenceListener)");
    	
        presenceListeners.add(listener);
        // Do a bulk announce for new listener if there are any nodes already here.
        if (localNodes.size()>0)
            listener.bulkNodeAdd(localNodes);        
    }
    
    /**
     * Removes a presence listener.
     * @param listener Presence listener.
     */
    public void removePresenceListener(PresenceListener listener)
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.removePresenceListener(PresenceListener)");
    	
        presenceListeners.remove(listener);
    }
    
    /**
     * Adds a new PresenceKey
     * @param presenceData PresenceKey to add
     */
    public void addPeer(PresenceKey presenceData)
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.addPeer(PresenceKey)");
    	
        if (presenceData.getPresenceName() != null){        	
            // Peer should be added if the data is complete
            synchronized(this){  
            	localNodes.put(presenceData.getPresenceNodeID(),presenceData);                
                //midasNodes.put(presenceData.getPresenceNodeID(),presenceData);
            }          
            announceNew(presenceData);
        }
        else
        	CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_WARNING, "NodePresence.addPeer(PresenceKey). " + 					
            		"Presence key name was null for: " + presenceData.getPresenceNodeID(), null);       
    }
    
    /**
     * Removes a PrecenceKey
     * @param presenceData PresenceKey to remove
     */
    public void removePeer(PresenceKey presenceData)
    {    	
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.removePeer(PresenceKey)");
    	
        if (localNodes.containsKey(presenceData.getPresenceNodeID()))
        {        	
            synchronized(this){
                localNodes.remove(presenceData.getPresenceNodeID());
                //midasNodes.remove(presenceData.getPresenceNodeID());
            }            
            sendMessage(announceOut,"off",presenceData);
            
            return;
        }
        /*synchronized(midasNodes){
            midasNodes.remove(presenceData.getPresenceNodeID());
        }*/
    }
    
    boolean scheduled = false;
    /**
     * Scheduled a bulk announce
     */
    public synchronized void scheduleBulkAnnounce()
    {    
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.scheduleBulkAnnounce()");
        java.util.Timer timer = new java.util.Timer(true);
        timer.schedule(new BulkTask(), bulkAnnounceDelay);
        scheduled = true;
    }
    
    /**
     * Perform a bulk announce
     */
    protected synchronized void bulkAnnounce()
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.bulkAnnounce()");
    	
        synchronized(this){
            scheduled = false;
        }
        CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "NodePresence.bulkAnnounce(). " +
        		"Sending bulk message of size: " + localNodes.size());
        if (localNodes.size() > 0)
        {    
            buff1.setLength(0);
            java.util.Iterator iter = localNodes.values().iterator();
            buff1 = new StringBuffer();
            while(iter.hasNext())
            {
                buff1.append(iter.next().toString());
                buff1.append("\n");
            }
            sendMessage(bulkOut,"bulk", buff1.toString());
        }
        else
        	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "NodePresence.bulkAnnounce(). " +            		
            		"There was nothing in the presence peer group list");               
    }
    /**
     * Start the announce pipe inbound
     */
    protected void startAnnouncePipe()
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.startAnnouncePipe()");
    	
        try{
            String type = PipeService.PropagateType;
            ID pipeID = (net.jxta.id.ID)MD5ID.createPipeID(peerGroup.getPeerGroupID(),"RDV diff", "status pipe");                        
            PipeAdvertisement pipeAdvertisement = AdvertisementUtilities.createPipeAdvertisement(pipeID, type);
            pipeAdvertisement.setName("announceIn");
            announceIn = peerGroup.getPipeService().createInputPipe(pipeAdvertisement, new InputListener());
            announceOut = peerGroup.getPipeService().createOutputPipe(pipeAdvertisement, 0);
            try {
            	peerGroup.getDiscoveryService().publish(pipeAdvertisement);
            	peerGroup.getDiscoveryService().remotePublish(pipeAdvertisement);
            } catch(Exception e){
            	CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_ERROR, "NodePresence.startAnnouncePipe(). " + 
    					"Error publishing our own advertisement: '" + e.getMessage() + "'", null);                
                throw new RuntimeException("Critical Error pubishing our own advertisement");
            }
        }catch(Exception ioe){
            ioe.printStackTrace();
        }
    }
    
    /**
     * Start the bulk announce pipe inbound
     */
    protected void startBulkPipeIn()
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.startBulkPipeIn()");
    	
        try{
            bulkIn = peerGroup.getPipeService().createInputPipe(bulkPipeAdvertisement, new InputListener());
        }catch(Exception ioe){
            ioe.printStackTrace();
        }
    }
    
    /**
     * Start the bulk pipe out
     */
    protected void startBulkPipeOut()
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.startBulkPipeOut()");
    	
        try{
            bulkOut = peerGroup.getPipeService().createOutputPipe(bulkPipeAdvertisement, 0);
        }catch(Exception ioe){
            ioe.printStackTrace();
        }
    }
    
    /**
     *  Given a JXTA ID node string, return the node Name
     * @param id JXTA node ID to map
     * @return Peer name
     */
    protected String idToName(String id ) 
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.idToName(" + id + ")");
    	
        Enumeration res;
        
        try {
            res = peerGroup.getDiscoveryService().getLocalAdvertisements(DiscoveryService.PEER, "PID", id);
            if ((null == res) || (!res.hasMoreElements()))
                return "";
        } catch (Exception e) {
            e.printStackTrace();
            return "";
        }
        
        while (res.hasMoreElements()) 
        {
            PeerAdvertisement peer=(PeerAdvertisement)res.nextElement();
            return peer.getName();
        }
        return "";
    }
    
    /**
     * Send a PresenceKey on a pipe with a specific tag
     * @param pipe Pipe to use
     * @param tag Tag to associate with the value
     * @param presenceData Value to send
     */
    protected void sendMessage(OutputPipe pipe, String tag, PresenceKey presenceData)
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.sendMessage(OutputPipe, " + tag + ", PresenceKey)");
    	
        try{
            Message message = new Message();
            message.addMessageElement( new net.jxta.endpoint.StringMessageElement(tag, presenceData.toString(), null, null));            
            sendMessage(pipe, message);
        }catch(Exception e){
        	CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_ERROR, "NodePresence.sendMessage(OutputPipe, " + tag + ", PresenceKey). " + 
        			"Error sending message: '" + e.getMessage() + "'", null);            
        }
    }
    
    /**
     * Send a message on a pipe with a specific tag
     * @param pipe Pipe to use
     * @param tag Tag to associate with the value
     * @param messageValue Message to send
     */
    protected void sendMessage(OutputPipe pipe, String tag, String messageValue)
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.sendMessage(OutputPipe, " + tag + ", " + messageValue + ")");
    	
        try{
            Message message = new Message();
            net.jxta.endpoint.StringMessageElement stringElement = new net.jxta.endpoint.StringMessageElement(tag, messageValue, null, null);
            message.addMessageElement(stringElement);
            sendMessage(pipe,message);
        }catch(Exception e){
        	CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_ERROR, "NodePresence.sendMessage(OutputPipe, " + tag + ", " + messageValue + "). " + 
        			"Error sending message: '" + e.getMessage() + "'", null);
        }
    }
    
    /**
     * Send a message on on the pipe
     * @param pipe Pipe to use
     * @param message Message to send
     */
    protected void sendMessage(OutputPipe pipe, Message message)
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.sendMessage(OutputPipe, Message");
    	
        try{
            while (!pipe.send(message)){
            }
        }catch(Exception e){
        	CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_ERROR, "NodePresence.sendMessage(OutputPipe, Message). " + 
        			"Error sending message: '" + e.getMessage() + "'", null);
        }
    }
    
    /**
     * Announce that the node is being added to the MIDAS network
     * @param presenceData Announces a new node associated with a PresenceKey
     */
    public void announceNew(PresenceKey presenceData)
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.announceNew(PresenceKey)");
    	
        sendMessage(announceOut,"on",presenceData);
    }
    
    /**
     * Announce that the node is leaving the network
     * @param presenceData PresenceKey of the node leaving the MIDAS network
     */
    public void announceLeave(PresenceKey presenceData)
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.announceLeave(PresenceKey)");
    	
        sendMessage(announceOut,"off",presenceData);
    }
    
    
    class EventIsolation extends Thread
    {
        PipeMsgEvent pipeMsgEvent;
        public EventIsolation(PipeMsgEvent pipeMsgEvent) 
        {
        	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "EventIsolation(PipeMsgEvent)");
        	
            this.pipeMsgEvent = pipeMsgEvent;
        }
        
        public void run()
        {
        	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "EventIsolation.run(). Session got a message");            
        	
            Message message = pipeMsgEvent.getMessage();
            //String rawData= null;
            MessageElement element = (MessageElement)message.getMessageElement(null, "on");
            if (element != null) 
            {
                buff1.setLength(0);
                String data = element.toString();
                PresenceKey presenceKey = PresenceKey.create(data);
                //if (peerGroup.getRendezVousService().isRendezVous())
                	localNodes.put(presenceKey.getPresenceNodeID(),presenceKey);
                //midasNodes.put(presenceKey.getPresenceNodeID(),presenceKey);             
                
                // Restart the bulk pipe if we were offline.
                if (thisNodeID.equals(presenceKey.getPresenceNodeID()) && bulkIn == null)
                    startBulkPipeIn();
                
                for (int i=0;i<presenceListeners.size();i++)
                {
                    PresenceListener presenceListener = (PresenceListener)presenceListeners.get(i);
                    presenceListener.nodeAdded(presenceKey);
                }
                return;
            }
            else
            {
                buff1.setLength(0);
                element = (MessageElement)message.getMessageElement(null, "off");
                if(element != null)
                {
                    String data = element.toString();
                    PresenceKey presenceData = PresenceKey.create(data);
                    localNodes.remove(presenceData.getPresenceNodeID());
                    for (int i=0;i<presenceListeners.size();i++)
                    {
                        PresenceListener presenceListener = (PresenceListener)presenceListeners.get(i);
                        presenceListener.nodeRemoved(presenceData);
                    }
                    return;
                }
                else
                {
                    buff1.setLength(0);
                    element = (MessageElement)message.getMessageElement(null, "bulk");
                    if(element != null)
                    {
                        String data = element.toString();
                        int start = 0;
                        int index = data.indexOf('\n',start);
                        if (index == -1)
                        {
                        	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "EventIsolation.run(). " +                         			
                            		"Empty bulk message sent");
                            return;
                        }
                        String list[] =data.split("\n");
                        for (int i =0;i < list.length;i++)
                        {
                            PresenceKey presenceKey = PresenceKey.create(list[i]);
                            localNodes.put(presenceKey.getPresenceNodeID(),presenceKey);
                        }
                        CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "EventIsolation.run(). " +
                        		"Received + " + list.length + " nodes in bulk message");
                        // Close the inbound pipe because it is no longer required
                        synchronized(this)
                        {
                            if (bulkIn != null)
                            {
                                bulkIn.close();
                                bulkIn =null;
                            }
                        }
                        for (int i=0;i<presenceListeners.size();i++)
                        {
                            PresenceListener presenceListener = (PresenceListener)presenceListeners.get(i);
                            presenceListener.bulkNodeAdd(localNodes);
                        }
                        return;
                    }
                }
            }            
        }// end run()
    }// End of class EventIsolation
    
    /** 
     * Listens to pipe messages 
     */
    protected class InputListener implements PipeMsgListener
    {
        /**
         * Pipe message event callback used to process inbound messages.
         * @param pipeMsgEvent Message event from pipe
         */
        public void pipeMsgEvent(PipeMsgEvent pipeMsgEvent) 
        {    
        	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "InputListener.pipeMsgEvent(PipeMsgEvent)");
        	
            if(headless)            	
                new EventIsolation(pipeMsgEvent).start();
            else
            {
                try {                    
                    javax.swing.SwingUtilities.invokeLater(new EventIsolation(pipeMsgEvent));
                }catch(Exception e){
                    // If headless then recall as a normal thread
                    headless = true;
                    pipeMsgEvent(pipeMsgEvent);
                }
            }
        }// end of pipeMsgEvent()        
    }// end of inner class InputListener
    
    /**
     * Rendezvous Listener. This looks at the events for nodes going on and offline
     * and announces the results to connected nodes.
     */
    protected class RDVListener implements RendezvousListener
    {
        /**
         * Rendezvous event processor callback. This looks at the events for nodes 
         * going on and offline and announces the results to connected nodes.
         * @param event Rendezvous event.
         */
        public void rendezvousEvent(RendezvousEvent event)
        {
        	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "RDVListener.rendezvousEvent(RendezvousEvent)");
        	
            PresenceKey presenceKey = new PresenceKey(event.getPeer());
            if (localNodes.containsKey(presenceKey.getPresenceNodeID()))
            {
                PresenceKey presenceKey2 = (PresenceKey)localNodes.get(presenceKey.getPresenceNodeID());               
                presenceKey = presenceKey2;               
                if (presenceKey.getPresenceName() == null)
                    presenceKey.setPresenceName("unnkown");    
            }
            switch (event.getType()) {
                case RendezvousEvent.RDVCONNECT:
                	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "RDVListener.rendezvousEvent(RendezvousEvent) ." +
                			"      RDVCONNECT:  " + event.getPeer());
                    if (peerGroup.getRendezVousService().isRendezVous())
                    {
                        addPeer(presenceKey);
                        scheduleBulkAnnounce();
                    }
                    break;
                case RendezvousEvent.RDVRECONNECT:
                	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "RDVListener.rendezvousEvent(RendezvousEvent) ." +
                			"      RDVRECONNECT:  " + event.getPeer());                	
                    if (peerGroup.getRendezVousService().isRendezVous())
                    {
                        addPeer(presenceKey);
                        scheduleBulkAnnounce();
                    }
                    break;
                case RendezvousEvent.RDVDISCONNECT:
                	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "RDVListener.rendezvousEvent(RendezvousEvent) ." +
                			"      RDVDISCONNECT:  " + event.getPeer());                	
                    if (peerGroup.getRendezVousService().isRendezVous())
                        removePeer(presenceKey);                                            
                    break;
                case RendezvousEvent.RDVFAILED:
                	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "RDVListener.rendezvousEvent(RendezvousEvent) ." +
                			"      RDVFAILED:  " + event.getPeer());                	
                    if (peerGroup.getRendezVousService().isRendezVous())
                        removePeer(presenceKey);                                            
                    break;
                case RendezvousEvent.CLIENTCONNECT:
                	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "RDVListener.rendezvousEvent(RendezvousEvent) ." +
                			"      CLIENTCONNECT:  " + event.getPeer());                	
                    if (peerGroup.getRendezVousService().isRendezVous())
                    {
                        addPeer(presenceKey);
                        scheduleBulkAnnounce();
                    }
                    break;
                case RendezvousEvent.CLIENTRECONNECT:
                	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "RDVListener.rendezvousEvent(RendezvousEvent) ." +
                			"      CLIENTRECONNECT:  " + event.getPeer());                	
                    if (peerGroup.getRendezVousService().isRendezVous())
                    {
                        addPeer(presenceKey);
                        scheduleBulkAnnounce();
                    }
                    break;
                case RendezvousEvent.CLIENTDISCONNECT:
                	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "RDVListener.rendezvousEvent(RendezvousEvent) ." +
                			"      CLIENTDISCONNECT:  " + event.getPeer());                	
                    if (peerGroup.getRendezVousService().isRendezVous())
                        removePeer(presenceKey);                    
                    break;
                case RendezvousEvent.CLIENTFAILED:
                	CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "RDVListener.rendezvousEvent(RendezvousEvent) ." +
                			"      CLIENTFAILED:  " + event.getPeer());                	
                    if (peerGroup.getRendezVousService().isRendezVous()
                    		&& presenceKey.getProxyName().equals(localPresenceKey.getPresenceName()))
                        removePeer(presenceKey);
                    break;
                default:
                    break;
            }
        }
    }// end of RendezvousListener
    
    /**
     * Bulk Task thread.
     */
    protected class BulkTask extends TimerTask
    {       
        /**
         * Run the presence thread
         */
        public void run() 
        {
        	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "BulkTask.run()");
            bulkAnnounce();
        }       
    }
    
    public PresenceKey getLocalPresenceKey()
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.getLocalPresenceKey()");
    	
    	return localPresenceKey;
    }
    
    public void nodeCall(String jxtaNodeId)
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "NodePresence.nodeCall(" + jxtaNodeId + ")");
    	
    	//Actualizamos la tabla del ultimo polling
    	PresenceKey pk = (PresenceKey)localNodes.get(jxtaNodeId);
    	if (pk != null)
    		pk.setLastChanged(System.currentTimeMillis());    	
    }
    
    public void notifyDissapearedNodes()
    {
    	PresenceKey pk = null;
    	long actualTime = System.currentTimeMillis();
    	Iterator it = localNodes.values().iterator();
    	Vector nodesToBeRemoved = new Vector();
    	while (it.hasNext()) 
    	{
    		pk = (PresenceKey)it.next();
    		if (pk.getLastChanged() != 0 && (pk.getLastChanged() + 20000) < actualTime)    		
    			nodesToBeRemoved.add(pk);
    	}
    	for (int i=0; i<nodesToBeRemoved.size(); i++)
			removePeer((PresenceKey)nodesToBeRemoved.get(i));
    }
}
